package com.moxiao.yfjh.module.system.mq.kafka;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonObject;
import com.moxiao.yfjh.module.system.dal.kafka.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.DescribeClusterResult;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ExecutionException;

//@Slf4j : 注解在类上, 为类提供一个属性名为 log 的 log4j 的日志对象
@Component
@Slf4j
public class KafkaProducer {

    /**
     * 用户注册：topics
     */
    public final static String KAFKA_BLIND_MEMBER_USER_KEY = "blind_member_user_register";


    private final KafkaTemplate<String, String> kafkaTemplate;
    /**
     * 用户头像认证：topics
     */
    public final static String KAFKA_BLIND_MEMBER_USER_AUTH = "blind_member_user_auth";
    /**
     * 用户身份证认证：topics
     */
    public final static String KAFKA_BLIND_MEMBER_USER_ID_CARD_AUTH = "blind_member_user_id_card_auth";
    /**
     * 内心独白：topics
     */
    public final static String KAFKA_BLIND_MEMBER_USER_INNER_MONOLOGUE = "blind_member_user_inner_monologue";
    /**
     * 打招呼：topics
     */
    public final static String KAFKA_BLIND_SAY_HELLO = "blind_member_user_say_hello";
    /**
     * 用户昵称：topics
     */
    public final static String KAFKA_BLIND_NICK_NAME = "blind_member_user_nick_name";

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;


    private Gson gson = new GsonBuilder().create();

    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void send(String msg) {
        Message message = new Message();

        message.setId(System.currentTimeMillis());
        message.setMsg(msg);
        message.setSendTime(new Date());
        log.info("【生产者发送消息：{}】", gson.toJson(message));
        //对 topic = hello2 的发送消息
        kafkaTemplate.send("hello2", gson.toJson(message));
    }

    /**
     * 发送用户注册消息
     */
    public void addSendLonginMember(JsonObject jsonObject) {
        Message message = new Message();
        // 增加唯一id
        jsonObject.addProperty("uid", UUID.randomUUID().toString().replace("-", ""));
        message.setId(System.currentTimeMillis());
        message.setMsg(jsonObject.toString());
        message.setSendTime(new Date());
        log.info("【生产者发送会员用户注册消息：{}】", gson.toJson(message));
        kafkaTemplate.send(KAFKA_BLIND_MEMBER_USER_KEY, gson.toJson(jsonObject));
    }

    /**
     * 发送头像实名认证消息
     */
    public void addSendAuthAvatar(JsonObject jsonObject) {
        Message message = new Message();
        // 增加唯一id
        jsonObject.addProperty("uid", UUID.randomUUID().toString().replace("-", ""));
        message.setId(System.currentTimeMillis());
        message.setMsg(jsonObject.toString());
        message.setSendTime(new Date());
        log.info("【生产者发送会员头像实名认证消息：{}】", gson.toJson(message));
        kafkaTemplate.send(KAFKA_BLIND_MEMBER_USER_AUTH, gson.toJson(jsonObject));
    }

    /**
     * 会员身份证认证
     */
    public void addSendInitFaceVerify(JsonObject jsonObject) {
        Message message = new Message();
        // 增加唯一id
        jsonObject.addProperty("uid", UUID.randomUUID().toString().replace("-", ""));
        message.setId(System.currentTimeMillis());
        message.setMsg(jsonObject.toString());
        message.setSendTime(new Date());
        log.info("【生产者发送会员身份认证消息：{}】", gson.toJson(message));
        kafkaTemplate.send(KAFKA_BLIND_MEMBER_USER_ID_CARD_AUTH, gson.toJson(jsonObject));
    }

    /**
     * 内心独白阿里云审核
     */
    public void addSendGreenTextScan(JsonObject jsonObject) {
        // 增加唯一id
        jsonObject.addProperty("uid", UUID.randomUUID().toString().replace("-", ""));
        Message message = new Message();
        message.setId(System.currentTimeMillis());
        message.setMsg(jsonObject.toString());
        message.setSendTime(new Date());
        log.info("【生产者发送会员修改内心独白消息：{}】", gson.toJson(message));
        kafkaTemplate.send(KAFKA_BLIND_MEMBER_USER_INNER_MONOLOGUE, gson.toJson(jsonObject));
    }

    public void addSendSayHello(JsonObject jsonObject) {
        // 增加唯一id
        jsonObject.addProperty("uid", UUID.randomUUID().toString().replace("-", ""));
        Message message = new Message();
        message.setId(System.currentTimeMillis());
        message.setMsg(jsonObject.toString());
        message.setSendTime(new Date());
        log.info("【生产者发送打招呼消息：{}】", gson.toJson(message));
        kafkaTemplate.send(KAFKA_BLIND_SAY_HELLO, gson.toJson(jsonObject));
    }

    /**
     * 昵称验证消息发送
     */
    public void addSendNickName(JsonObject jsonObject) {
        // 增加唯一id
        jsonObject.addProperty("uid", UUID.randomUUID().toString().replace("-", ""));
        Message message = new Message();
        message.setId(System.currentTimeMillis());
        message.setMsg(jsonObject.toString());
        message.setSendTime(new Date());
        log.error("【生产者发送用户昵称消息：{}】", gson.toJson(message));
        kafkaTemplate.send(KAFKA_BLIND_NICK_NAME, gson.toJson(jsonObject));
    }

    /**
     * kafka 链接状态测试
     */
    public boolean isKafkaConnected() {
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        try (AdminClient adminClient = AdminClient.create(properties)) {
            DescribeClusterResult describeClusterResult = adminClient.describeCluster();
            //如果连接失败，这将触发异常
            describeClusterResult.nodes().get();
            // 如果我们到达这里，连接就成功了
            log.info("kafka链接正常！");
            return true;
        } catch (InterruptedException | ExecutionException e) {
            // 连接失败
            log.info("kafka链接失败！{}", e.toString());
            return false;
        }
    }
}
